package com.min.kafka.service;

import cn.hutool.json.JSONUtil;
import cn.hutool.log.Log;
import com.min.kafka.SendCallBack;
import com.min.kafka.entity.KMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * kafka 发送消息处理类
 * @author: fengt
 * @date: 2021/4/1 18:13
 **/
public class KafkaService {

    @Autowired
    private KafkaTemplate<Object, Object> kafkaTemplate;

    final Log log = Log.get();

    /**
     * 发送消息
     * @param message 消息体
     * @param topic 主题
     * @return: org.springframework.util.concurrent.ListenableFuture
     * @author: fengt
     * @date: 2021/4/1 16:32
     **/
    public void send(String topic, KMessage message) {
        send(topic, message, null);
    }

    /**
     * 发送消息
     * @param message 消息体
     * @param topic 主题
     * @param cb 成功或失败回调函数
     * @return: void
     * @author: fengt
     * @date: 2021/4/23 10:44
     **/
    public void send(String topic, KMessage message, SendCallBack cb) {
        message.setCount(message.getCount() + 1);
        ListenableFuture<SendResult<Object, Object>> future = kafkaTemplate.send(topic, message);
        future.addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() {
            @Override
            public void onSuccess(SendResult<Object, Object> objectObjectSendResult) {
                log.debug("{} 发送成功，结果：{}", topic, JSONUtil.toJsonStr(objectObjectSendResult));
                if (cb != null) {
                    cb.successToDo(objectObjectSendResult);
                }
            }

            @Override
            public void onFailure(Throwable throwable) {
                log.error(topic + " 发送失败", topic, throwable);
                if (cb != null) {
                    cb.failureToDo(throwable);
                }
            }

        });
    }

}
